/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.restapi.controllers;

import eu.dicodeproject.analysis.restapi.SchemaConfig;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;

import java.io.IOException;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.WritableByteArrayComparable;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @author jakob
 */
public abstract class FilteringController extends ControllerBase {

  //TODO FIXME this needs re-factoring

  private byte[] textQualifier;
  private byte[] langQualifier;

  protected Scan getScan(String start, String stop, String language, String regex, JSONObject params)
      throws IOException, JSONException {
    return getScan(this.columnFamily, this.textQualifier, this.langQualifier, start, stop, language, regex, params);
  }

  protected Scan getScan(SchemaConfig schemaConfig, String start, String stop, String language, String regex, JSONObject params)
      throws IOException, JSONException {

    //TODO FIXME this is a hack: fix in the schema, or find a nicer way to configure
    byte[] langColumn = schemaConfig.languageColumn();
    if (language != null && language.equals("de") && schemaConfig.deLanguageColumn() != null) {
      langColumn = schemaConfig.deLanguageColumn();
    }

    return getScan(schemaConfig.columnFamily(), schemaConfig.textColumn(), langColumn, start, stop, language, regex, params);
  }

  protected Scan getScan(byte[] hBaseColumnFamily, byte[] textQualifier, byte[] langQualifier, String start, String stop, String language, String regex, JSONObject params)
      throws IOException, JSONException {
    // set HBase Scan
    Scan scan = new Scan();
    scan.addFamily(hBaseColumnFamily);
    scan.addFamily(Bytes.toBytes("f")); //FIXME: move to configuration (List of column families in schema config????) 

    // assuming that row key starts with creation date
    if (start != null) {
      start = cleanDate(start);
      scan.setStartRow(Bytes.toBytes(start));
      params.put("start", start);
    }
    if (stop != null) {
      stop = cleanDate(stop);
      scan.setStopRow(Bytes.toBytes(stop));
      params.put("stop", stop);
    }

    // filter based on language and regex
    FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
    if (language != null) {
      byte[] l = Bytes.toBytes(language);
      Filter langFilter = new SingleColumnValueFilter(hBaseColumnFamily, langQualifier, CompareFilter.CompareOp.EQUAL, l);
      filters.addFilter(langFilter);
      params.put("language", language);
    }
    if (regex != null) {
      WritableByteArrayComparable r = new RegexStringComparator(regex);
      Filter regexFilter = new SingleColumnValueFilter(hBaseColumnFamily, textQualifier, CompareFilter.CompareOp.EQUAL, r);
      filters.addFilter(regexFilter);
      params.put("regex", regex);
    }
    scan.setFilter(filters);

    return scan;
  }


  private String cleanDate(String dateString) {
    return dateString.replaceAll("[^\\d_]", "");
  }

}
